"""
消息相关API路由
"""
from typing import List, Dict, Any, Optional
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel, Field
import asyncio
from datetime import datetime
import orjson
import re

from app.core.kafka_client import get_kafka_client
from app.utils.logger import get_logger

router = APIRouter()
logger = get_logger(__name__)


class KafkaMessage(BaseModel):
    """Kafka消息模型"""
    topic: str
    partition: int
    offset: int
    timestamp: Optional[int]
    key: Optional[str]
    value: Any
    headers: Dict[str, Any] = {}


class SendMessageRequest(BaseModel):
    """发送消息请求"""
    topic: str
    value: Any
    key: Optional[str] = None
    partition: Optional[int] = None


class SendMessageResponse(BaseModel):
    """发送消息响应"""
    success: bool
    message: str


class MessageFilters(BaseModel):
    """消息过滤器"""
    partition: Optional[int] = Field(None, description="分区号")
    start_time: Optional[datetime] = Field(None, description="开始时间")
    end_time: Optional[datetime] = Field(None, description="结束时间")
    key_pattern: Optional[str] = Field(None, description="Key匹配模式")
    value_pattern: Optional[str] = Field(None, description="Value搜索内容")
    min_offset: Optional[int] = Field(None, description="最小偏移量")
    max_offset: Optional[int] = Field(None, description="最大偏移量")
    use_regex: bool = Field(False, description="是否使用正则表达式")
    case_sensitive: bool = Field(False, description="是否区分大小写")


class PaginatedMessagesResponse(BaseModel):
    """分页消息响应"""
    messages: List[KafkaMessage]
    total_count: int
    page: int
    page_size: int
    total_pages: int
    has_next: bool
    has_prev: bool


@router.get("/{topic_name}", response_model=List[KafkaMessage])
async def get_messages(
    topic_name: str,
    max_messages: int = Query(default=100, ge=1, le=10000, description="最大消息数量"),
    load_mode: str = Query(default="latest", description="加载模式: latest(最新), all(全量), earliest(从头开始)"),
    from_beginning: bool = Query(default=False, description="是否从头开始读取")
):
    """
    获取Topic消息

    参数:
    - topic_name: Topic名称
    - max_messages: 最大消息数量 (1-10000)
    - load_mode: 加载模式
      - latest: 获取最新的N条消息
      - all: 获取全量消息（忽略max_messages限制）
      - earliest: 从最早的消息开始获取N条
    - from_beginning: 是否从头开始读取（优先级低于load_mode）
    """
    try:
        kafka_client = get_kafka_client()

        # 检查Topic是否存在
        topics = await kafka_client.list_topics()
        if topic_name not in topics:
            raise HTTPException(status_code=404, detail=f"Topic不存在: {topic_name}")

        # 根据加载模式调整参数
        if load_mode == "all":
            # 全量模式：设置一个很大的数字，支持海量数据
            actual_max_messages = 10000000  # 1000万条消息
            consume_from_beginning = True
        elif load_mode == "earliest":
            # 从头开始模式
            actual_max_messages = max_messages
            consume_from_beginning = True
        else:  # latest
            # 最新消息模式
            actual_max_messages = max_messages
            consume_from_beginning = from_beginning

        logger.info(f"获取消息: topic={topic_name}, mode={load_mode}, max={actual_max_messages}, from_beginning={consume_from_beginning}")

        messages = []
        async for message in kafka_client.consume_messages(
            topic_name,
            actual_max_messages,
            from_beginning=consume_from_beginning
        ):
            messages.append(KafkaMessage(**message))

        logger.info(f"成功获取 {len(messages)} 条消息")
        return messages

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取消息失败: {e}")
        raise HTTPException(status_code=500, detail=f"获取消息失败: {str(e)}")


@router.post("/send", response_model=SendMessageResponse)
async def send_message(request: SendMessageRequest):
    """发送消息"""
    try:
        kafka_client = get_kafka_client()
        
        # 检查Topic是否存在
        topics = await kafka_client.list_topics()
        if request.topic not in topics:
            raise HTTPException(status_code=404, detail=f"Topic不存在: {request.topic}")
        
        # 发送消息
        success = await kafka_client.send_message(
            topic=request.topic,
            value=request.value,
            key=request.key,
            partition=request.partition
        )
        
        if success:
            return SendMessageResponse(
                success=True,
                message="消息发送成功"
            )
        else:
            raise HTTPException(status_code=500, detail="消息发送失败")
            
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"发送消息失败: {e}")
        raise HTTPException(status_code=500, detail=f"发送消息失败: {str(e)}")


@router.get("/{topic_name}/paginated", response_model=PaginatedMessagesResponse)
async def get_messages_paginated(
    topic_name: str,
    page: int = Query(default=1, ge=1, description="页码"),
    page_size: int = Query(default=100, ge=1, le=1000, description="每页大小"),
    partition: Optional[int] = Query(default=None, description="分区号"),
    start_time: Optional[str] = Query(default=None, description="开始时间 (ISO格式)"),
    end_time: Optional[str] = Query(default=None, description="结束时间 (ISO格式)"),
    key_pattern: Optional[str] = Query(default=None, description="Key匹配模式"),
    value_pattern: Optional[str] = Query(default=None, description="Value搜索内容"),
    min_offset: Optional[int] = Query(default=None, description="最小偏移量"),
    max_offset: Optional[int] = Query(default=None, description="最大偏移量"),
    use_regex: bool = Query(default=False, description="是否使用正则表达式"),
    case_sensitive: bool = Query(default=False, description="是否区分大小写"),
    load_mode: str = Query(default="latest", description="加载模式: latest, earliest")
):
    """
    分页获取Topic消息，支持高级过滤

    参数:
    - topic_name: Topic名称
    - page: 页码 (从1开始)
    - page_size: 每页大小 (1-1000)
    - partition: 指定分区号
    - start_time/end_time: 时间范围过滤
    - key_pattern: Key匹配模式 (支持通配符 *)
    - value_pattern: Value内容搜索
    - min_offset/max_offset: 偏移量范围
    - use_regex: 是否使用正则表达式匹配
    - case_sensitive: 是否区分大小写
    - load_mode: latest(从最新开始) 或 earliest(从最早开始)
    """
    try:
        kafka_client = get_kafka_client()

        # 检查Topic是否存在
        topics = await kafka_client.list_topics()
        if topic_name not in topics:
            raise HTTPException(status_code=404, detail=f"Topic不存在: {topic_name}")

        # 构建过滤器
        filters = MessageFilters(
            partition=partition,
            start_time=datetime.fromisoformat(start_time.replace('Z', '')) if start_time else None,
            end_time=datetime.fromisoformat(end_time.replace('Z', '')) if end_time else None,
            key_pattern=key_pattern,
            value_pattern=value_pattern,
            min_offset=min_offset,
            max_offset=max_offset,
            use_regex=use_regex,
            case_sensitive=case_sensitive
        )

        logger.info(f"分页获取消息: topic={topic_name}, page={page}, size={page_size}, mode={load_mode}")

        # 获取消息 (这里简化实现，实际应该支持真正的分页)
        all_messages = []
        max_fetch = page * page_size + 1000  # 多获取一些用于过滤

        async for message in kafka_client.consume_messages(
            topic_name,
            max_fetch,
            from_beginning=(load_mode == "earliest")
        ):
            # 应用过滤器
            if _apply_message_filters(message, filters):
                all_messages.append(KafkaMessage(**message))

        # 计算分页
        total_count = len(all_messages)
        total_pages = (total_count + page_size - 1) // page_size
        start_idx = (page - 1) * page_size
        end_idx = start_idx + page_size

        paginated_messages = all_messages[start_idx:end_idx]

        logger.info(f"分页结果: 总数={total_count}, 当前页={page}, 返回={len(paginated_messages)}")

        return PaginatedMessagesResponse(
            messages=paginated_messages,
            total_count=total_count,
            page=page,
            page_size=page_size,
            total_pages=total_pages,
            has_next=page < total_pages,
            has_prev=page > 1
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"分页获取消息失败: {e}")
        raise HTTPException(status_code=500, detail=f"分页获取消息失败: {str(e)}")


def _apply_message_filters(message: Dict[str, Any], filters: MessageFilters) -> bool:
    """应用消息过滤器"""
    try:
        # 分区过滤
        if filters.partition is not None and message.get('partition') != filters.partition:
            return False

        # 时间范围过滤
        if filters.start_time or filters.end_time:
            msg_timestamp = message.get('timestamp')
            if msg_timestamp:
                msg_time = datetime.fromtimestamp(msg_timestamp / 1000)  # Kafka时间戳是毫秒
                if filters.start_time and msg_time < filters.start_time:
                    return False
                if filters.end_time and msg_time > filters.end_time:
                    return False

        # 偏移量范围过滤
        msg_offset = message.get('offset', 0)
        if filters.min_offset is not None and msg_offset < filters.min_offset:
            return False
        if filters.max_offset is not None and msg_offset > filters.max_offset:
            return False

        # Key模式匹配
        if filters.key_pattern:
            msg_key = message.get('key', '')
            if not msg_key:
                return False

            if filters.use_regex:
                pattern = filters.key_pattern
                flags = 0 if filters.case_sensitive else re.IGNORECASE
                if not re.search(pattern, msg_key, flags):
                    return False
            else:
                # 通配符匹配
                pattern = filters.key_pattern.replace('*', '.*')
                flags = 0 if filters.case_sensitive else re.IGNORECASE
                if not re.match(pattern, msg_key, flags):
                    return False

        # Value内容搜索
        if filters.value_pattern:
            msg_value = str(message.get('value', ''))
            search_text = filters.value_pattern

            if not filters.case_sensitive:
                msg_value = msg_value.lower()
                search_text = search_text.lower()

            if filters.use_regex:
                flags = 0 if filters.case_sensitive else re.IGNORECASE
                if not re.search(filters.value_pattern, str(message.get('value', '')), flags):
                    return False
            else:
                if search_text not in msg_value:
                    return False

        return True

    except Exception as e:
        logger.warning(f"过滤器应用失败: {e}")
        return True  # 过滤失败时包含该消息


@router.get("/{topic_name}/search", response_model=List[KafkaMessage])
async def search_messages(
    topic_name: str,
    query: str = Query(..., description="搜索查询"),
    max_results: int = Query(default=100, ge=1, le=1000, description="最大结果数"),
    search_in: str = Query(default="value", description="搜索范围: key, value, both"),
    use_regex: bool = Query(default=False, description="是否使用正则表达式"),
    case_sensitive: bool = Query(default=False, description="是否区分大小写")
):
    """
    搜索Topic中的消息

    参数:
    - topic_name: Topic名称
    - query: 搜索查询字符串
    - max_results: 最大结果数
    - search_in: 搜索范围 (key, value, both)
    - use_regex: 是否使用正则表达式
    - case_sensitive: 是否区分大小写
    """
    try:
        kafka_client = get_kafka_client()

        # 检查Topic是否存在
        topics = await kafka_client.list_topics()
        if topic_name not in topics:
            raise HTTPException(status_code=404, detail=f"Topic不存在: {topic_name}")

        logger.info(f"搜索消息: topic={topic_name}, query='{query}', in={search_in}")

        results = []
        search_count = 0
        max_scan = 10000  # 最多扫描10000条消息

        async for message in kafka_client.consume_messages(topic_name, max_scan, from_beginning=False):
            if search_count >= max_scan:
                break

            search_count += 1

            # 执行搜索匹配
            if _match_search_query(message, query, search_in, use_regex, case_sensitive):
                results.append(KafkaMessage(**message))

                if len(results) >= max_results:
                    break

        logger.info(f"搜索完成: 扫描={search_count}, 匹配={len(results)}")
        return results

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"搜索消息失败: {e}")
        raise HTTPException(status_code=500, detail=f"搜索消息失败: {str(e)}")


def _match_search_query(
    message: Dict[str, Any],
    query: str,
    search_in: str,
    use_regex: bool,
    case_sensitive: bool
) -> bool:
    """匹配搜索查询"""
    try:
        search_targets = []

        if search_in in ['key', 'both']:
            key = message.get('key', '')
            if key:
                search_targets.append(str(key))

        if search_in in ['value', 'both']:
            value = message.get('value', '')
            if value:
                search_targets.append(str(value))

        if not search_targets:
            return False

        for target in search_targets:
            if use_regex:
                flags = 0 if case_sensitive else re.IGNORECASE
                if re.search(query, target, flags):
                    return True
            else:
                if not case_sensitive:
                    target = target.lower()
                    query = query.lower()
                if query in target:
                    return True

        return False

    except Exception as e:
        logger.warning(f"搜索匹配失败: {e}")
        return False
